package com.example.pro1;

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;

/**
 * @program: reactivePro
 * @description: 传统编码方式的问题
 * @author: zy
 * @create: 2025-04-14 11:35
 */
public class CarFactory {
    // 安装车门, 耗时 2 秒
    static String installDoor(String car) {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装车门");
        return car + "+车门";
    }
    // 安装轮子, 耗时 1.5 秒
    static String installWheel(String car) {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装轮子");
        return car + "+轮子";
    }
    // 安装座椅, 耗时 2.5 秒
    static String installSeat(String car) {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装座椅");
        return car + "+座椅";
    }

    /** 传统单线程:  主线程被阻塞 6 秒, 线程利用率低, 也就是前述的 Thread-Per-Request 问题 */
    static void makeCar1() {
        String car = "汽车";
        car = installDoor(car);
        car = installWheel(car);
        car = installSeat(car);
        System.out.println("makeCar1 完成! " + car);
    }

    /** 主线程现在不阻塞了,但是
     新的问题: 子线程被阻塞 6 秒, 子线程利用率低, 之前的问题转嫁到子线程上 */
    static void makeCar2() {
        new Thread(()->{
            String car = "汽车";
            car = installDoor(car);
            car = installWheel(car);
            car = installSeat(car);
            System.out.println("makeCar2 完成! " + car);
        }).start();
    }

    /** 没有线程阻塞问题, 但是结果输出: makeCar3 完成! 汽车+座椅
     新的问题: 1.太多的线程, 线程利用率在总体上仍然不高; 2.线程通信和协调问题, 导致结果不正确
     */
    static void makeCar3() throws Exception {
        String[] car = {"汽车"};
        new Thread(()->car[0] = installDoor(car[0])).start();
        new Thread(()->car[0] = installWheel(car[0])).start();
        new Thread(()->car[0] = installSeat(car[0])).start();
        Thread.sleep(3000);
        System.out.println("makeCar3 完成! " + car[0]);
        // 结果输出: makeCar3 完成! 汽车+座椅
    }



    // 对三个业务方法进行封装, 方法的第二个参数是一个回调函数接口,
// 用于在业务方法执行结束之后执行后续业务方法
    static void installDoor(String car, Consumer<String> consumer){
        CompletableFuture<String> future = CompletableFuture.supplyAsync(   ()->{
            return installDoor(car);
        });
        future.thenAcceptAsync(consumer);
    }
    static void installWheel(String car, Consumer<String> consumer){
        CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
            return installWheel(car);
        });
        future.thenAcceptAsync(consumer);
    }
    static void installSeat(String car, Void v){
        String ret = installSeat(car);
        System.out.println("makeCar4 完成!  "+ ret);
    }
    //解决了线程通信和协调问题, 但是
    //新的问题: ”回调地狱“问题, 编码难度大和可读性差
    static void makeCar4() {
        installDoor("汽车",a->{
            installWheel(a,b->{
                installSeat(b, null);
            });
        });
    }



    // 基于 Reactor 框架的解决方案
    static void makeCar5() throws IOException {
        Mono.just("汽车")  //初始值

                //任务发布者, 它是一个线程池, 用于发布任务, 它是reactor内置线程池
                .publishOn(Schedulers.parallel())   // 切换到并行线程池,它是reactor内置线程池
                //任务执行
                .map(car->installDoor(car))
                .map(car->installWheel(car))
                .map(car->installSeat(car))    //这三个操作是并行的吗
                //订阅者
                .subscribe((car)->System.out.println("makeCar5 完成! " + car));
        // 阻止线程结束, 在此等待
        System.in.read();
    }

    //改进上一个方案:
    static void makeCar6() throws InterruptedException {
        var latch = new CountDownLatch(1);  // 用于等待任务完成, 用于替换上面的System.in.read();　更可靠地等待异步任务完成
        Mono.just("汽车")
                .subscribeOn(Schedulers.boundedElastic())  // 适用于阻塞操作，　比 parallel() 更适合 IO 密集型任务
                .map(CarFactory::installDoor)
                .map(CarFactory::installWheel)
                .map(CarFactory::installSeat)
                .subscribe(
                        car -> {
                            System.out.println("makeCar6 完成! " + car);
                            latch.countDown();  // 任务完成，释放锁
                        },
                        error -> {
                            System.err.println("发生错误: " + error.getMessage());
                            latch.countDown();
                        }
                );
        latch.await();  // 等待异步任务完成
    }


    public static void main(String[] args) throws Exception {
       // makeCar1();
      //  makeCar2();
     //   makeCar3();
     //   makeCar4();
     // makeCar5();
        makeCar6();
        System.out.println("aaa");
        System.in.read();
    }


}
